package org.weishe.pagerank.hbase;

import java.io.IOException;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;

public class EndReducer extends TableReducer<DoubleWritable, Text, ImmutableBytesWritable> {

	public static final byte[] CF = "cf1".getBytes();
	public static final byte[] WHO = "who".getBytes();
	public static final byte[] WEIGHT = "weight".getBytes();

	@Override
	public void reduce(DoubleWritable key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {

		for (Text t : values) {
			// 保证在hbase中式倒序输出
			String rowKey = (1000000 - key.get()) + t.toString();
			Put put = new Put(Bytes.toBytes(rowKey));
			put.add(CF, WEIGHT, key.toString().getBytes());
			put.add(CF, WHO, t.getBytes());
			context.write(null, put);
		}

	}

}
